#!/usr/bin/python

# Node View
# visualize average CPU (mem, I/O) usage of specific job on each cluster machine
# compared with average CPU (mem, I/O) usage per machine

# CPU usage : %
# rbytes: KB/s
# wbytes: KB/s
# virtual mem: MB
# physical mem: MB

import sys
import re

def getvalue(a):
	return a[a.find("=")+1:].strip("\"")

def getjobid(a):
	A = a.split('_')
	return A[2]

if len(sys.argv) <= 3:
	print "./COMMAND [task tracker log file name] [min time] [max time]?\n"
	sys.exit(-1)

mint = int(sys.argv[2])
print mint
maxt = int(sys.argv[3])
print maxt
# scan task tracker log once
# find out how many MP job are running
# and also record Host, Tasktracker and DataNode information
jobs = {}
tasktracker = {}
datanode = {}
host = {}
for l in open(sys.argv[1]):
	A = l.split()
	id = 0
	type = {}
	time = 0
	rbytes = 0.0
	wbytes = 0.0
	msize = 0.0
	rss = 0.0
	cpu = 0.0
	for a in A:
		# check type
		if a.startswith("property"):
			B = (getvalue(a)).split(".")
			if len(B) == 1:
				# host or job
				if B[0].startswith("host:"):
					# host
					type = "host"
				else:
					# job
					id = getjobid(B[0])
					if not jobs.has_key(id):
						jobs[id]=0
						print "find job "+id
			elif len(B) == 5:
				# tasktracker
				type = "tasktracker"
			else:
				# datanode
				type = "datanode"

		if a.startswith("walltime"):
			time = int(getvalue(a))
		if a.startswith("readbytes_thput="):
			rbytes = float(getvalue(a))
		if a.startswith("writebytes_thput="):
			wbytes = float(getvalue(a))
		if a.startswith("VmSize"):
			msize = float(getvalue(a))
		if a.startswith("VmRSS"):
			rss = float(getvalue(a))
		if a.startswith("cpu_per"):
			cpu = float(getvalue(a))
			if cpu > 8.0:
				cpu = 0.0

	# skip job information
	if id != 0:
		continue

	# skip info out of time range
	if (time < mint) or (time > maxt):
		continue

	if type == "host":
		if not host.has_key(time):
			host[time] = {}
			host[time]["rbytes"] = 0.0 
			host[time]["wbytes"] = 0.0
			host[time]["msize"] = 0.0
			host[time]["rss"] = 0.0
			host[time]["cpu"] = 0.0
		host[time]["rbytes"] += (rbytes/1024)
		host[time]["wbytes"] += (wbytes/1024)
		host[time]["msize"] += (msize/1024)
		host[time]["rss"] += (rss/1024)
		host[time]["cpu"] += (cpu*100)

	if type == "tasktracker":
		if not tasktracker.has_key(time):
			tasktracker[time] = {}
			tasktracker[time]["rbytes"] = 0.0 
			tasktracker[time]["wbytes"] = 0.0
			tasktracker[time]["msize"] = 0.0
			tasktracker[time]["rss"] = 0.0
			tasktracker[time]["cpu"] = 0.0
		tasktracker[time]["rbytes"] += (rbytes/1024)
		tasktracker[time]["wbytes"] += (wbytes/1024)
		tasktracker[time]["msize"] += (msize/1024)
		tasktracker[time]["rss"] += (rss/1024)
		tasktracker[time]["cpu"] += (cpu*100)

	if type == "datanode":
		if not datanode.has_key(time):
			datanode[time] = {}
			datanode[time]["rbytes"] = 0.0 
			datanode[time]["wbytes"] = 0.0
			datanode[time]["msize"] = 0.0
			datanode[time]["rss"] = 0.0
			datanode[time]["cpu"] = 0.0
		datanode[time]["rbytes"] += (rbytes/1024)
		datanode[time]["wbytes"] += (wbytes/1024)
		datanode[time]["msize"] += (msize/1024)
		datanode[time]["rss"] += (rss/1024)
		datanode[time]["cpu"] += (cpu*100)

mf = open('host.dat', 'w')
for k,v in host.items():
	mf.write(str(k-mint-0.7)+' '+ str(v["rbytes"])+' '+ str(v["wbytes"])+' '+ str(v["msize"])+' '+ str(v["rss"]) +' '+ str(v["cpu"])+'\n' )
	mf.write(str(k-mint)+' '+ str(v["rbytes"])+' '+ str(v["wbytes"])+' '+ str(v["msize"])+' '+ str(v["rss"]) +' '+ str(v["cpu"])+'\n' )
	mf.write(str(k-mint+0.7)+' '+ str(v["rbytes"])+' '+ str(v["wbytes"])+' '+ str(v["msize"])+' '+ str(v["rss"]) +' '+ str(v["cpu"])+'\n' )
	# clear host info
	v["rbytes"]=0.0
	v["wbytes"]=0.0
	v["msize"]=0.0
	v["rss"]=0.0
	v["cpu"]=0.0
mf.close()



# for each MP job
# scan log and record job information
lastid = {}
for jobid in jobs.keys():
	for l in open(sys.argv[1]):
		A = l.split()
		id = 0
		type = {}
		time = 0
		rbytes = 0.0
		wbytes = 0.0
		msize = 0.0
		rss = 0.0
		cpu = 0.0
		for a in A:
			# get job id if exists
			if a.startswith("property"):
				B = (getvalue(a)).split(".")
				if (len(B) == 1) and (B[0].startswith("attempt")):
					id = getjobid(B[0])
				
			if a.startswith("walltime"):
				time = int(getvalue(a))
			if a.startswith("readbytes_thput="):
				rbytes = float(getvalue(a))
			if a.startswith("writebytes_thput="):
				wbytes = float(getvalue(a))
			if a.startswith("VmSize"):
				msize = float(getvalue(a))
			if a.startswith("VmRSS"):
				rss = float(getvalue(a))
			if a.startswith("cpu_per"):
				cpu = float(getvalue(a))
				if cpu > 8.0:
					cpu = 0.0

		# only process the correct job information
		if (id == 0) or (id != jobid) :
			continue

		# skip info out of time range
		if (time < mint) or (time > maxt):
			continue

		host[time]["rbytes"] += (rbytes/1024)
		host[time]["wbytes"] += (wbytes/1024)
		host[time]["msize"] += (msize/1024)
		host[time]["rss"] += (rss/1024)
		host[time]["cpu"] += (cpu*100)

	mf = open("job"+jobid+".dat", 'w')
	print "job"+jobid+".dat"
	lastid = jobid
	for k,v in host.items():
		mf.write(str(k-mint-0.7)+' '+ str(v["rbytes"])+' '+ str(v["wbytes"])+' '+ str(v["msize"])+' '+ str(v["rss"]) +' '+ str(v["cpu"])+'\n' )
		mf.write(str(k-mint)+' '+ str(v["rbytes"])+' '+ str(v["wbytes"])+' '+ str(v["msize"])+' '+ str(v["rss"]) +' '+ str(v["cpu"])+'\n' )
		mf.write(str(k-mint+0.7)+' '+ str(v["rbytes"])+' '+ str(v["wbytes"])+' '+ str(v["msize"])+' '+ str(v["rss"]) +' '+ str(v["cpu"])+'\n' )
	mf.close()

# add tasktracker info 
for k,v in tasktracker.items():
	v["rbytes"] += host[k]["rbytes"]
	v["wbytes"] += host[k]["wbytes"]
	v["msize"] += host[k]["msize"]
	v["rss"] += host[k]["rss"]
	v["cpu"] += host[k]["cpu"]

mf = open('tasktracker.dat', 'w')
print "tasktracker.dat"
for k,v in tasktracker.items():
	mf.write(str(k-mint-0.7)+' '+ str(v["rbytes"])+' '+ str(v["wbytes"])+' '+ str(v["msize"])+' '+ str(v["rss"]) +' '+ str(v["cpu"])+'\n' )
	mf.write(str(k-mint)+' '+ str(v["rbytes"])+' '+ str(v["wbytes"])+' '+ str(v["msize"])+' '+ str(v["rss"]) +' '+ str(v["cpu"])+'\n' )
	mf.write(str(k-mint+0.7)+' '+ str(v["rbytes"])+' '+ str(v["wbytes"])+' '+ str(v["msize"])+' '+ str(v["rss"]) +' '+ str(v["cpu"])+'\n' )
mf.close()

# add datanode info
for k,v in datanode.items():
	v["rbytes"] += tasktracker[k]["rbytes"]
	v["wbytes"] += tasktracker[k]["wbytes"]
	v["msize"] += tasktracker[k]["msize"]
	v["rss"] += tasktracker[k]["rss"]
	v["cpu"] += tasktracker[k]["cpu"]

mf = open('datanode.dat', 'w')
print "datanode.dat"
for k,v in datanode.items():
	mf.write(str(k-mint-0.7)+' '+ str(v["rbytes"])+' '+ str(v["wbytes"])+' '+ str(v["msize"])+' '+ str(v["rss"]) +' '+ str(v["cpu"])+'\n' )
	mf.write(str(k-mint)+' '+ str(v["rbytes"])+' '+ str(v["wbytes"])+' '+ str(v["msize"])+' '+ str(v["rss"]) +' '+ str(v["cpu"])+'\n' )
	mf.write(str(k-mint+0.7)+' '+ str(v["rbytes"])+' '+ str(v["wbytes"])+' '+ str(v["msize"])+' '+ str(v["rss"]) +' '+ str(v["cpu"])+'\n' )
mf.close()

